package drds.server.route.shard_algorithm;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.List;

/**
 * 例子 按月份列分区 ，每个自然月一个分片，格式 between操作解析的范例
 */
public class ByMonth extends AbstractShardAlgorithm implements ShardAlgorithm {
	private String dateFormatString;
	private ThreadLocal<SimpleDateFormat> dateFormat;
	//
	private String beginDateString;
	private String endDateString;
	//
	private Calendar beginDateCalendar;
	private Calendar endDateCalendar;
	private int shardNum;

	@Override
	public void init() {
		try {
			beginDateCalendar = Calendar.getInstance();
			beginDateCalendar.setTime(new SimpleDateFormat(dateFormatString).parse(beginDateString));
			dateFormat = new ThreadLocal<SimpleDateFormat>() {
				@Override
				protected SimpleDateFormat initialValue() {
					return new SimpleDateFormat(dateFormatString);
				}
			};
			if (endDateString != null && !endDateString.equals("")) {
				endDateCalendar = Calendar.getInstance();
				endDateCalendar.setTime(new SimpleDateFormat(dateFormatString).parse(endDateString));
				shardNum = ((endDateCalendar.get(Calendar.YEAR) - beginDateCalendar.get(Calendar.YEAR)) * 12 + endDateCalendar.get(Calendar.MONTH) - beginDateCalendar.get(Calendar.MONTH)) + 1;

				if (shardNum <= 0) {
					throw new java.lang.IllegalArgumentException("Incorrect time range for month partitioning!");
				}
			} else {
				shardNum = -1;
			}
		} catch (ParseException e) {
			throw new java.lang.IllegalArgumentException(e);
		}
	}

	/**
	 * For circulatory partition, calculated value of target partition needs to
	 * be rotated to fit the partition range
	 */
	private int fixShardIndex(int shardIndex) {
		/**
		 * If target date is previous of start time of partition setting, shift
		 * the delta range between target and start date to be positive value
		 */
		if (shardIndex < 0) {
			shardIndex = shardNum - (-shardIndex) % shardNum;
		}

		if (shardIndex >= shardNum) {
			shardIndex = shardIndex % shardNum;
		}

		return shardIndex;
	}

	@Override
	public Integer calculate(String columnValue) {
		try {
			int targetPartition;
			Calendar calendar = Calendar.getInstance();
			calendar.setTime(dateFormat.get().parse(columnValue));
			targetPartition = ((calendar.get(Calendar.YEAR) - beginDateCalendar.get(Calendar.YEAR)) * 12 + calendar.get(Calendar.MONTH) - beginDateCalendar.get(Calendar.MONTH));

			/**
			 * For circulatory partition, calculated value of target partition
			 * needs to be rotated to fit the partition range
			 */
			if (shardNum > 0) {
				targetPartition = fixShardIndex(targetPartition);
			}
			return targetPartition;

		} catch (ParseException e) {
			throw new IllegalArgumentException(new StringBuilder().append("columnValue:").append(columnValue).append(" Please check if the format satisfied.").toString(), e);
		}
	}

	@Override
	public Integer[] calculateRange(String beginColumnValue, String endColumnValue) {
		try {

			Calendar calendar = Calendar.getInstance();
			SimpleDateFormat dateFormat = new SimpleDateFormat(dateFormatString);
			calendar.setTime(dateFormat.parse(beginColumnValue));
			int startShardIndex = ((calendar.get(Calendar.YEAR) - beginDateCalendar.get(Calendar.YEAR)) * 12 + calendar.get(Calendar.MONTH) - beginDateCalendar.get(Calendar.MONTH));

			calendar.setTime(dateFormat.parse(endColumnValue));
			int endShardIndex = ((calendar.get(Calendar.YEAR) - beginDateCalendar.get(Calendar.YEAR)) * 12 + calendar.get(Calendar.MONTH) - beginDateCalendar.get(Calendar.MONTH));

			List<Integer> nodeIndexList = new ArrayList<Integer>();

			while (startShardIndex <= endShardIndex) {
				Integer nodeIndex = fixShardIndex(startShardIndex);
				if (Collections.frequency(nodeIndexList, nodeIndex) < 1) {
					nodeIndexList.add(nodeIndex);
				}
				startShardIndex++;
			}
			int size = nodeIndexList.size();
			return (nodeIndexList.toArray(new Integer[size]));
		} catch (ParseException e) {
			return new Integer[0];
		}
	}

	@Override
	public int getShardNum() {
		int shardNum = this.shardNum;
		return shardNum;
	}

	public void setDateFormatString(String dateFormatString) {
		this.dateFormatString = dateFormatString;
	}

	public void setBeginDateDateString(String beginDateString) {
		this.beginDateString = beginDateString;
	}

	public void setEndDateString(String endDateString) {
		this.endDateString = endDateString;
	}

}
